# @Version        : 1.0
# @Update Time    : 2025/10/13 22:44
# @File           : errors.py
# @IDE            : PyCharm
# @Desc           : 文件描述信息

from typing import Any, Dict, Optional, Sequence, Union
from authx.core.enums import TokenErrorType, ErrorCodeEnum


__all__ = [
    "AuthError",
    "TokenError",
    "MalformedToken",
    "SignatureInvalid",
    "TokenExpired",
    "RefreshExpired",
    "SessionError",
    "SessionRevoked",
    "SessionNotFound",
    "UserBanned",
    "PermissionException",
    "RoleException",
    "PolicyViolation",
    "InputError",
    "ServiceError",
]


class AuthError(Exception):
    """认证/授权领域异常基类（与 Web/HTTP 解耦）

    Attributes:
        code: 机器可读的错误码（字符串），用于语义标识与上层映射
        error_code: 统一数值错误码（来自 ErrorCodeEnum），便于前端/客户端一致处理
        message: 人类可读的错误信息（开发/日志友好）
        context: 结构化上下文（便于日志/追踪），例如 {"jti": "...", "user_id": "..."}
    """

    def __init__(
        self,
        code: str,
        message: str,
        *,
        context: Optional[Dict[str, Any]] = None,
        cause: Optional[BaseException] = None,
        error_code: Optional[int] = None,
    ) -> None:
        super().__init__(message)
        self.code = code
        self.message = message
        self.context = context or {}
        # 记录底层原因；真正的异常链建议在 raise 时用 `from e`
        self.cause = cause
        # 统一的数值错误码（可选）
        self.error_code = error_code

    def to_dict(self) -> Dict[str, Any]:
        """序列化为结构化字典（便于向上层传递/记录）"""
        data: Dict[str, Any] = {
            "code": self.code,
            "message": self.message,
            "context": self.context or {},
        }
        if self.error_code is not None:
            data["error_code"] = self.error_code
        return data

    def __str__(self) -> str:
        return f"{self.code}: {self.message}"

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}(code={self.code!r}, message={self.message!r}, context={self.context!r})"


# ---------------- Token 相关 ----------------


class TokenError(AuthError):
    """Token 相关错误基类

    支持直接传入 TokenErrorType（或其 error_type 字符串），自动从枚举中获取 message，并规范化上下文中的 error_type 字段。
    """

    def __init__(
        self,
        message: Optional[str] = None,
        *,
        error_type: Optional[Union[TokenErrorType, str]] = None,
        context: Optional[Dict[str, Any]] = None,
        cause: Optional[BaseException] = None,
    ) -> None:
        ctx: Dict[str, Any] = dict(context or {})
        resolved_message = message

        if error_type is not None:
            if isinstance(error_type, TokenErrorType):
                # 规范化上下文中的错误类型为字符串
                ctx.setdefault("error_type", error_type.error_type)
                # 如果未指定 message，则从枚举获取
                resolved_message = resolved_message or error_type.message
            else:
                # 传入的是字符串类型的 error_type
                ctx.setdefault("error_type", str(error_type))
                # message 保持不变（可能由上层明确传入）

        if not resolved_message:
            resolved_message = "令牌错误"

        super().__init__("token_error", resolved_message, context=ctx, cause=cause)


class MalformedToken(TokenError):
    """令牌格式/结构不合法（解析失败、字段缺失等）"""

    def __init__(
        self,
        reason: str = "令牌格式不合法",
        *,
        context: Optional[Dict[str, Any]] = None,
        cause: Optional[BaseException] = None,
    ) -> None:
        super().__init__(reason, context=context, cause=cause)
        self.code = "token_malformed"
        self.error_code = ErrorCodeEnum.TOKEN_INVALID.code


class SignatureInvalid(TokenError):
    """令牌签名不合法/密钥不匹配"""

    def __init__(
        self,
        reason: str = "令牌签名无效",
        *,
        context: Optional[Dict[str, Any]] = None,
        cause: Optional[BaseException] = None,
    ) -> None:
        super().__init__(reason, context=context, cause=cause)
        self.code = "token_signature_invalid"
        self.error_code = ErrorCodeEnum.TOKEN_INVALID.code


class TokenExpired(TokenError):
    """访问令牌过期"""

    def __init__(
        self,
        token_type: str = "access",
        exp: Optional[int] = None,
        *,
        context: Optional[Dict[str, Any]] = None,
    ) -> None:
        msg = f"{token_type} token 已过期"
        ctx = dict(context or {})
        if exp is not None:
            ctx["exp"] = exp
        super().__init__(msg, context=ctx)
        self.code = "token_expired"
        self.token_type = token_type
        self.exp = exp
        self.error_code = ErrorCodeEnum.TOKEN_EXPIRED.code


class RefreshExpired(TokenError):
    """刷新令牌过期或刷新窗口关闭"""

    def __init__(
        self,
        reason: str = "refresh 令牌过期或窗口已关闭",
        *,
        context: Optional[Dict[str, Any]] = None,
    ) -> None:
        super().__init__(reason, context=context)
        self.code = "refresh_expired"
        self.error_code = ErrorCodeEnum.TOKEN_EXPIRED.code


# ---------------- Session 会话相关 ----------------


class SessionError(AuthError):
    """会话相关错误基类"""

    def __init__(
        self, message: str = "会话错误", *, context: Optional[Dict[str, Any]] = None
    ) -> None:
        super().__init__("session_error", message, context=context)


class SessionRevoked(SessionError):
    """会话已撤销（注销、踢下线、封禁联动等）"""

    def __init__(
        self, reason: str = "会话已撤销", *, context: Optional[Dict[str, Any]] = None
    ) -> None:
        super().__init__(reason, context=context)
        self.code = "session_revoked"


class SessionNotFound(SessionError):
    """会话不存在"""

    def __init__(self, *, context: Optional[Dict[str, Any]] = None) -> None:
        super().__init__("会话不存在", context=context)
        self.code = "session_not_found"
        self.error_code = ErrorCodeEnum.TOKEN_NOT_FOUND.code


# ---------------- 用户禁用/封禁 ----------------


class UserBanned(AuthError):
    """用户被封禁（可带截止时间与原因）"""

    def __init__(
        self,
        until_ts: Optional[int],
        reason: Optional[str] = None,
        *,
        context: Optional[Dict[str, Any]] = None,
    ) -> None:
        msg = "用户被封禁" + (f"，截至 {until_ts}" if until_ts else "")
        if reason:
            msg += f"（{reason}）"
        ctx = dict(context or {})
        if until_ts is not None:
            ctx["until_ts"] = until_ts
        if reason:
            ctx["ban_reason"] = reason
        super().__init__("user_banned", msg, context=ctx)
        self.until_ts = until_ts
        self.reason = reason
        self.error_code = ErrorCodeEnum.ACCOUNT_DISABLED.code


# ---------------- 授权（角色/权限） ----------------


class PermissionException(AuthError):
    """权限不足"""

    def __init__(
        self,
        login_type: str,
        required: Sequence[str],
        owned: Optional[Sequence[str]] = None,
        *,
        context: Optional[Dict[str, Any]] = None,
    ) -> None:
        req_set = set(required or [])
        own_set = set(owned or [])
        missing = sorted(list(req_set - own_set))
        msg = f"权限不足，缺少: {', '.join(missing) if missing else '未知'}"
        ctx = dict(context or {})
        ctx.update(
            {
                "login_type": login_type,
                "required": list(req_set),
                "owned": list(own_set),
                "missing": missing,
            }
        )
        super().__init__("permission_denied", msg, context=ctx)
        self.required = list(req_set)
        self.owned = list(own_set)
        self.missing = missing
        self.error_code = ErrorCodeEnum.NO_PERMISSION.code


class RoleException(AuthError):
    """角色不足"""

    def __init__(
        self,
        login_type: str,
        required: Sequence[str],
        owned: Optional[Sequence[str]] = None,
        *,
        context: Optional[Dict[str, Any]] = None,
    ) -> None:
        req_set = set(required or [])
        own_set = set(owned or [])
        missing = sorted(list(req_set - own_set))
        msg = f"角色不足，缺少: {', '.join(missing) if missing else '未知'}"
        ctx = dict(context or {})
        ctx.update(
            {
                "login_type": login_type,
                "required": list(req_set),
                "owned": list(own_set),
                "missing": missing,
            }
        )
        super().__init__("role_denied", msg, context=ctx)
        self.error_code = ErrorCodeEnum.NO_ROLE.code


# ---------------- 策略/配置 ----------------


class PolicyViolation(AuthError):
    """登录并发策略/配置约束违反（通常属于实现问题或非法操作）"""

    def __init__(
        self, message: str = "策略约束违反", *, context: Optional[Dict[str, Any]] = None
    ) -> None:
        super().__init__("policy_violation", message, context=context)
        self.error_code = ErrorCodeEnum.BUSINESS_ERROR.code


# ---------------- 输入参数/服务错误 ----------------


class InputError(AuthError):
    """输入参数错误（通常映射为 400 Bad Request）"""

    def __init__(
        self, message: str = "参数错误", *, context: Optional[Dict[str, Any]] = None
    ) -> None:
        super().__init__("parameter_error", message, context=context)
        self.error_code = ErrorCodeEnum.PARAMETER_ERROR.code


class ServiceError(AuthError):
    """服务内部错误（通常映射为 500 Internal Server Error）"""

    def __init__(
        self,
        message: str = "服务内部错误",
        *,
        context: Optional[Dict[str, Any]] = None,
        cause: Optional[BaseException] = None,
    ) -> None:
        super().__init__("service_error", message, context=context, cause=cause)
        self.error_code = ErrorCodeEnum.SYSTEM_ERROR.code
